package beautiful.butterfly.distributed_task_processing_engine.task_manage.task_data_items_processor;

import beautiful.butterfly.distributed_task_processing_engine.api.MultipleTaskDataItemCollectorAndExecuterAndExecuter;
import beautiful.butterfly.distributed_task_processing_engine.api.SingleTaskDataItemCollectorAndExecuterAndExecuter;
import beautiful.butterfly.distributed_task_processing_engine.api.TaskDataItemCollectorAndExecuter;
import beautiful.butterfly.distributed_task_processing_engine.task_manage.task_data_items_processor.task_data_item_data_manager.TaskItemStringAndParameterString;
import beautiful.butterfly.distributed_task_processing_engine.task_manage.task_processor.Statistics;
import beautiful.butterfly.distributed_task_processing_engine.task_manage.task_processor.TaskProcessor;
import beautiful.butterfly.distributed_task_processing_engine.user_definition.TaskUserDefinition;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.Array;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;


@Slf4j
public class ForEverRunTaskDataItemProcessor<TaskDataItem> implements ITaskDataItemProcessor, Runnable
{

    protected TaskProcessor taskProcessor;
    protected TaskDataItemCollectorAndExecuter<TaskDataItem> taskDataItemCollectorAndExecuter;
    protected List<TaskDataItem> taskDataItemList = new CopyOnWriteArrayList<TaskDataItem>();
    /**
     * 正在处理中的任务队列
     */
    protected List<Object> runningTaskDataItemList = new CopyOnWriteArrayList<Object>();
    /**
     * 在重新取数据,可能会有重复的数据.在重新取数据前,从runningTaskDataItemList拷贝得来
     */
    protected List<TaskDataItem> maybeRepeatedTaskDataItemList = new CopyOnWriteArrayList<TaskDataItem>();
    List<Thread> threadList = new CopyOnWriteArrayList<Thread>();
    TaskUserDefinition taskUserDefinition;
    Comparator<TaskDataItem> taskItemComparator;
    Statistics statistics;
    Lock runningTaskDataItemListLock = new ReentrantLock();
    Lock fetchMultipleTaskLock = new ReentrantLock();
    Lock fetchTaskItemsLock = new ReentrantLock();
    /**
     * 是否可以批处理
     */
    volatile boolean isMultipleTask = false;

    /**
     * 是否已经获得终止调度信号
     */
    volatile boolean stopSchedule = false;// 用户停止队列调度
    volatile boolean isSleeping = false;

    /**
     * 创建一个调度处理器
     */

    public ForEverRunTaskDataItemProcessor(//
                                           TaskProcessor taskProcessor,//
                                           TaskDataItemCollectorAndExecuter<TaskDataItem> taskDataItemCollectorAndExecuter,//
                                           Statistics statistics) throws Exception//
    {
        this.taskProcessor = taskProcessor;
        this.taskUserDefinition = this.taskProcessor.getTaskUserDefinition();
        //
        this.taskDataItemCollectorAndExecuter = taskDataItemCollectorAndExecuter;
        this.taskItemComparator = new TaskComparator(this.taskDataItemCollectorAndExecuter.getComparator());
        if (this.taskDataItemCollectorAndExecuter instanceof SingleTaskDataItemCollectorAndExecuterAndExecuter<?>)
        {
            if (taskUserDefinition.getMaxExecuteNumber() > 1)
            {
                taskUserDefinition.setMaxExecuteNumber(1);
            }
            isMultipleTask = false;
        } else
        {
            isMultipleTask = true;
        }
        if (taskUserDefinition.getFetchNumber() > taskUserDefinition.getThreadNumber() * 10)
        {

            log.warn("参数设置不合理，系统性能不佳。【每次从数据库获取的数量fetchnum】 >= 【线程数量threadnum】 *【最少循环次数10】 ");
        }
        //
        this.statistics = statistics;
        //
        for (int i = 0; i < taskUserDefinition.getThreadNumber(); i++)
        {
            this.startThread(i);//一个任务处理器开启几个任务项线程
        }
    }

    class TaskComparator implements Comparator<TaskDataItem>
    {
        Comparator<TaskDataItem> comparator;

        public TaskComparator(Comparator<TaskDataItem> comparator)
        {
            this.comparator = comparator;
        }

        public int compare(TaskDataItem leftTask, TaskDataItem rightTask)
        {
            statistics.addOtherCompareCount(1);
            return this.comparator.compare(leftTask, rightTask);
        }

        public boolean equals(Object object)
        {
            return this.comparator.equals(object);
        }
    }

    //
    private void startThread(int index)
    {
        Thread thread = new Thread(this);//共享信息
        threadList.add(thread);
        String threadName = this.taskProcessor.getTaskProcessorInfo().getTaskId() + "-" + this.taskProcessor.getCurrentThreadNo() + "-exe" + index;
        thread.setName(threadName);
        thread.start();
    }

    //
    public void clearTaskDataItemList()
    {
        this.taskDataItemList.clear();
    }

    public boolean runningTaskItemListIsAllFinshed()
    {
        return this.taskDataItemList.size() == 0 //
                &&//
                this.runningTaskDataItemList.size() == 0;//
    }

    public boolean isSleeping()
    {
        return this.isSleeping;
    }

    /**
     * 需要注意的是，调度服务器从配置中心注销的工作，必须在所有线程退出的情况下才能做
     */
    public void setStopScheduleFlagAndClearTaskList() throws Exception
    { // 设置停止调度的标志,调度线程发现这个标志，执行完当前任务后，就退出调度
        this.stopSchedule = true;
        //清除所有未处理任务,但已经进入处理队列的，需要处理完毕
        this.taskDataItemList.clear();
    }


    //
    public void run()
    {
        long startTimeMillis = 0;
        Object taskDataItem = null;
        while (true)
        {
            try
            {
                // 停止队列调度
                if (this.stopSchedule == true)
                {
                    synchronized (this.threadList)
                    {
                        this.threadList.remove(Thread.currentThread());
                        if (this.threadList.size() == 0)
                        {
                            this.taskProcessor.unRegisterTaskProcessor();
                        }
                    }
                    return;
                }
                //
                if (!this.isMultipleTask)
                {
                    taskDataItem = this.getTaskDataItem();
                } else
                {
                    taskDataItem = this.getTaskDataItems();
                }
                if (taskDataItem == null)
                {
                    this.fetchTaskItems();//每个线程都可以调用处理程序,所以需要增加比较任务数据项的方法,判断是否处理重复.
                    continue;
                }
                //
                try
                {
                    this.runningTaskDataItemList.add(taskDataItem);
                    startTimeMillis = taskProcessor.taskAndTaskProcessorAndTaskDataItemManage.zooKeeperCurrentTimeMillis();
                    if (!this.isMultipleTask)
                    {//single_task
                        if (((SingleTaskDataItemCollectorAndExecuterAndExecuter<Object>) this.taskDataItemCollectorAndExecuter)//
                                .dealWith(taskDataItem))
                        {
                            addSuccessNum(1, taskProcessor.taskAndTaskProcessorAndTaskDataItemManage.zooKeeperCurrentTimeMillis()
                                    - startTimeMillis);
                        } else
                        {
                            addFailNum(1, taskProcessor.taskAndTaskProcessorAndTaskDataItemManage.zooKeeperCurrentTimeMillis()
                                    - startTimeMillis);
                        }
                    } else
                    {//multiple_task
                        if (((MultipleTaskDataItemCollectorAndExecuterAndExecuter<Object>) this.taskDataItemCollectorAndExecuter)//
                                .dealWith((Object[]) taskDataItem))
                        {
                            addSuccessNum(((Object[]) taskDataItem).length, taskProcessor.taskAndTaskProcessorAndTaskDataItemManage.zooKeeperCurrentTimeMillis()
                                    - startTimeMillis);
                        } else
                        {
                            addFailNum(((Object[]) taskDataItem).length, taskProcessor.taskAndTaskProcessorAndTaskDataItemManage.zooKeeperCurrentTimeMillis()
                                    - startTimeMillis);
                        }
                    }
                } catch (Throwable ex)
                {
                    if (this.isMultipleTask == false)
                    {
                        addFailNum(1, taskProcessor.taskAndTaskProcessorAndTaskDataItemManage.zooKeeperCurrentTimeMillis() - startTimeMillis);
                    } else
                    {
                        addFailNum(((Object[]) taskDataItem).length, taskProcessor.taskAndTaskProcessorAndTaskDataItemManage.zooKeeperCurrentTimeMillis()
                                - startTimeMillis);
                    }
                    log.error("TaskDataItem :" + taskDataItem + " ", ex);
                } finally
                {
                    this.runningTaskDataItemList.remove(taskDataItem);
                }
            } catch (Throwable e)
            {
                throw new RuntimeException(e);
            }
        }
    }

    /**
     * 获取单个任务,注意lock是必须,否则在maybeRepeatTaskList的数据处理上会出现冲突
     */
    public TaskDataItem getTaskDataItem()
    {
        runningTaskDataItemListLock.lock();
        try
        {
            TaskDataItem taskDataItem = null;
            while (true)
            {
                if (this.taskDataItemList.size() == 0)
                {
                    return null;
                } else
                {
                    taskDataItem = this.taskDataItemList.remove(0);
                    //删除其他线程正在处理的任务数据项
                    if (this.returnTaskDataItemIsInMaybeRepeatedTaskDataItemListAndRemoveTaskDataItemFromMaybeRepeatedTaskDataItemListIfTaskDataItemIsInMaybeRepeatedTaskDataItemList(taskDataItem))
                    {
                        //当前任务数据项不可用,继续下一个尝试,直到可用
                    } else
                    {
                        return taskDataItem;
                    }
                }
            }
        } finally
        {
            runningTaskDataItemListLock.unlock();
        }
    }


    @SuppressWarnings("unchecked")
    public TaskDataItem[] getTaskDataItems()
    {
        fetchMultipleTaskLock.lock();
        try
        {
            if (this.taskDataItemList.size() == 0)//0
            {
                return null;
            }
            //
            int minSizeBetweenTaskListSizeAndMaxExecuteNumber = taskDataItemList.size();
            if (taskUserDefinition.getMaxExecuteNumber() == 0)
            {//0
                //不做限制
            } else
            {
                minSizeBetweenTaskListSizeAndMaxExecuteNumber = taskDataItemList.size() > taskUserDefinition.getMaxExecuteNumber() ? taskUserDefinition.getMaxExecuteNumber() : taskDataItemList.size();//min
            }

            List<TaskDataItem> taskDataItemList = new ArrayList<TaskDataItem>();
            int count = 0;
            TaskDataItem taskDataItem = null;
            while (count < minSizeBetweenTaskListSizeAndMaxExecuteNumber && ((taskDataItem = this.getTaskDataItem()) != null))
            {
                taskDataItemList.add(taskDataItem);
                count = count + 1;
            }
            if (taskDataItemList.size() == 0)
            {
                return null;
            } else
            {
                return taskDataItemList.toArray((TaskDataItem[]) Array.newInstance(taskDataItemList.get(0).getClass(), 0));
            }
        } finally
        {
            fetchMultipleTaskLock.unlock();
        }
    }

    /**
     * 存在则返回true,且从这个防止重复的队列里面删除该任务数据项,其他情况都返回false
     */
    @SuppressWarnings("unchecked")
    protected boolean returnTaskDataItemIsInMaybeRepeatedTaskDataItemListAndRemoveTaskDataItemFromMaybeRepeatedTaskDataItemListIfTaskDataItemIsInMaybeRepeatedTaskDataItemList(TaskDataItem task)
    {
        //我们只关心maybeRepeatedTaskDataItemList与taskDataItemList关系中存在交集的部分的那些任务数据项.
        if (this.maybeRepeatedTaskDataItemList.size() == 0)
        {
            return false;
        } else
        {
            TaskDataItem[] taskDataItems = (TaskDataItem[]) this.maybeRepeatedTaskDataItemList.toArray();
            for (int i = 0; i < taskDataItems.length; i++)
            {
                //保证不会两个线程处理同一个任务数据项
                if (this.taskItemComparator.compare(task, taskDataItems[i]) == 0)
                {
                    this.maybeRepeatedTaskDataItemList.remove(taskDataItems[i]);
                    return true;
                }
            }
            return false;
        }

    }

    /**
     * 为什么这种情况会出现任务重复的情况:
     * 装载数据
     */
    protected int fetchTaskItems() throws InterruptedException
    {
        //进行同步处理,保证一个只有一个线程获取任务数据项.但是不保证出于绝对线程安全的情况,可能其lock块中的对象在其他方法中被使用.
        fetchTaskItemsLock.lock();
        try
        {   // 如果被其他的线程装载任务数据项,则直接返回
            if (this.taskDataItemList.size() > 0 || this.stopSchedule)
            {
                return this.taskDataItemList.size();
            }
            // 在每次数据处理完毕后休眠固定的时间
            if (this.taskUserDefinition.getSleepTimeIntervalBeforeFetchTasks() > 0)
            {
                this.isSleeping = true;
                TimeUnit.SECONDS.sleep(taskUserDefinition.getSleepTimeIntervalBeforeFetchTasks());
                this.isSleeping = false;
            }
            //任务数据项list中为空,但是不代表正在待处理的任务数据项list中为空
            putRunningTaskItemListToMaybeRepeatedTaskItemList();

            try
            {
                List<TaskItemStringAndParameterString> taskItemStringAndParameterStringList = this.taskProcessor.waiterunningTaskItemListIsAllFinshedAndSynchronizedGetTaskItemStringAndParameterStringList();
                if (taskItemStringAndParameterStringList.size() > 0)
                {
                    //底层的对象不可以被共享
                    List<TaskItemStringAndParameterString> newTaskItemStringAndParameterStringList = new ArrayList<TaskItemStringAndParameterString>();
                    synchronized (taskItemStringAndParameterStringList)
                    {
                        for (TaskItemStringAndParameterString taskItemStringAndParameterString : taskItemStringAndParameterStringList)
                        {
                            newTaskItemStringAndParameterStringList.add(taskItemStringAndParameterString);
                        }
                    }
                    List<TaskDataItem> taskDataItemList = this.taskDataItemCollectorAndExecuter.getTaskDataItemList(//
                            taskUserDefinition.getTaskParameter(),//
                            taskUserDefinition.getFetchNumber(),//
                            this.taskProcessor.getTaskItemCount(),//
                            newTaskItemStringAndParameterStringList //
                    );
                    taskProcessor.getTaskProcessorInfo().setLastFetchDataTimestamp(new Timestamp(taskProcessor.taskAndTaskProcessorAndTaskDataItemManage.zooKeeperCurrentTimeMillis()));
                    if (taskDataItemList != null)
                    {
                        this.taskDataItemList.addAll(taskDataItemList);
                    }
                }
                addFetchNum(taskDataItemList.size());
                if (taskDataItemList.size() <= 0)
                {
                    // 判断当没有数据的是否，是否需要退出调度
                    if (this.taskProcessor.continueWhenHasData())
                    {
                        if (taskUserDefinition.getSleepTimeWhenNotHasData() > 0)
                        {
                            this.isSleeping = true;
                            Thread.sleep(taskUserDefinition.getSleepTimeWhenNotHasData());
                            this.isSleeping = false;
                        }
                    }
                }
                return this.taskDataItemList.size();
            } catch (Throwable ex)
            {
                log.error("", ex);
            }
            return 0;
        } finally
        {
            fetchTaskItemsLock.unlock();
        }
    }

    /**
     * 将running队列的数据拷贝到可能重复的队列中
     * <pre>
     *     防止重复的模型是
     *     taskDataItem->瞬间->runningTaskDataItemList
     *
     *     如果这个时候再次向系统进行取任务数据项(new all),则可能去除的数据正在别处理.
     *     所以增加maybeRepeatedTaskDataItemList.当线程取出taskDataItem,需要到maybeRepeatedTaskDataItemList进行判断,如果存在则删除该项不再进行处理[maybeRepeatedTaskDataItemList和taskDataItemList]
     *     在这个过程中必须对maybeRepeatedTaskDataItemList加锁,保证不被两个线程重复的删除.
     *
     *     taskDataItem<-瞬间<-runningTaskDataItemList
     * </pre>
     */
    @SuppressWarnings("unchecked")
    public void putRunningTaskItemListToMaybeRepeatedTaskItemList()
    {
        runningTaskDataItemListLock.lock();
        try
        {
            this.maybeRepeatedTaskDataItemList.clear();//正在运行的任务存在可能被别人重复执行的可能
            if (this.runningTaskDataItemList.size() == 0)
            {
                return;
            }
            Object[] objects = this.runningTaskDataItemList.toArray();
            for (int i = 0; i < objects.length; i++)
            {
                if (!this.isMultipleTask)
                {
                    this.maybeRepeatedTaskDataItemList.add((TaskDataItem) objects[i]);
                } else
                {
                    TaskDataItem[] taskDataItems = (TaskDataItem[]) objects[i];
                    for (int j = 0; j < taskDataItems.length; j++)
                    {
                        this.maybeRepeatedTaskDataItemList.add(taskDataItems[j]);
                    }
                }
            }
        } finally
        {
            runningTaskDataItemListLock.unlock();
        }
    }

    //
    public void addFetchNum(long num)
    {
        this.statistics.addFetchDataCount(1);
        this.statistics.addTaskFetchNum(num);
    }

    public void addSuccessNum(long num, long spendTime)
    {
        this.statistics.addDealDataSucess(num);
        this.statistics.addDealSpendTime(spendTime);
    }

    public void addFailNum(long num, long spendTime)
    {
        this.statistics.addDealDataFail(num);
        this.statistics.addDealSpendTime(spendTime);
    }


}
